package vn.ebuzz.sms.worker;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.UnsupportedEncodingException;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLConnection;
import java.net.URLEncoder;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;

import org.apache.log4j.Logger;

import vn.ebuzz.sms.core.Configuration;
import vn.ebuzz.sms.core.Database;
import vn.ebuzz.sms.core.Incoming;

public class PushQueue {
	private static Logger log = Logger.getLogger("procqueue");
	private static PushQueue q = null;
	volatile boolean isRunning = true;

	private PushQueue(){
		PushWorker worker = new PushWorker();
		worker.start();
	}
	
	public static PushQueue getInstance(){
		if(q == null)
			q = new PushQueue();
		return q;
	}

	public void stop(){
		isRunning = false;
	}

	private class PushWorker extends Thread{
		@Override
		public void run() {
			long start = System.currentTimeMillis();
			long threadId = Thread.currentThread().getId();	
			while(isRunning){
				log.debug("PushWorker" + threadId + " check DB");
				Connection conn = null;
				Statement st = null;
				ResultSet rs = null;
				ArrayList<Incoming> list = new ArrayList<Incoming>();
				try{
					conn = Database.getConnection();
					st = conn.createStatement();

					rs = st.executeQuery("select id, sender, receiver," +
							" message from incoming" +
							" where status = " + Configuration.MO_QUEUED2PUSH +
							" order by created asc LIMIT " + Configuration.MAX_BATCH);
					while(rs.next()){
						Incoming mo = new Incoming();
						mo.setId(rs.getInt("id"));
						mo.setMessage(rs.getString("message"));
						mo.setReceiver(rs.getString("receiver"));
						mo.setSender(rs.getString("sender"));
						list.add(mo);
					}
				} catch(Exception e){
					log.error("Error", e);
				} finally{
					if(rs != null){
						try {
							rs.close();
						} catch (SQLException e) {
							log.error("Error", e);							
						}
					}
					if(st != null){
						try {
							st.close();
						} catch (SQLException e) {
							log.error("Error", e);
						}
					}
					if(conn != null) Database.closeConnection(conn);
				}
				if(list.size() > 0){
					String tmp = "";
					for(int i = 0; i < list.size(); i ++){
						tmp += list.get(i).getId();
						if(i < list.size() - 1)
							tmp += ",";
					}
					try{
						conn = Database.getConnection();
						st = conn.createStatement();
						String sql = "update incoming" +
						" set status = " + Configuration.MO_PUSHED +
						" where id in (" + tmp + ")";
						st.executeUpdate(sql);
					} catch (Exception e){
						log.error("Error", e);
					} finally {
						if(st != null){
							try {
								st.close();
							} catch (SQLException e) {
								log.error("Error", e);
							}
						}
						if(conn != null) Database.closeConnection(conn);
					}
					// push mo to CP in separated threads
					for(Incoming mo : list){
						ConnectionWorker cw = new ConnectionWorker(mo);
						cw.start();
					}
					long end = System.currentTimeMillis();
					long exectime = end - start;
					if(exectime < Configuration.THRESHOLD){
						try {
							Thread.sleep(exectime);
						} catch (InterruptedException e) {
							log.error("Error", e);
						}						
					}
				}
				else{
					try {
						Thread.sleep(Configuration.SLEEP);
					} catch (InterruptedException e) {
						log.error("Error", e);
					}					
				}
				log.debug("ProcessWorker" + threadId + " finish " + list.size() + " jobs");								
			}
		}
	}

	private class ConnectionWorker extends Thread{
		private Incoming incoming;
		public ConnectionWorker(Incoming incoming){
			this.incoming = incoming;
			log.info("ConnectionWorker: send " + incoming.getId());
		}

		@Override
		public void run() {
			try {
				String queryString = "http://mdoor.binhmuc.com/MDoor/receive" +
				"?id=" + incoming.getId() + 
				"&msisdn=" + incoming.getSender() + 
				"&shortcode=" + incoming.getReceiver() + 
				"&message=" + URLEncoder.encode(incoming.getMessage(), "UTF-8");
				log.debug("ConnectionWorker: push " + queryString);
				URL url = new URL(queryString);
				URLConnection conn = url.openConnection();
				conn.setConnectTimeout(5000);
				conn.setReadTimeout(5000);
				
				InputStream is = conn.getInputStream();
				BufferedReader reader = new BufferedReader(new InputStreamReader(is));
				
				StringBuilder sb = new StringBuilder();
				String line = "";
				while((line = reader.readLine()) != null){
					sb.append(line);
				}
				reader.close();
				is.close();
				
				log.info("ConnectionWorker: result " + sb.toString());
			} catch (UnsupportedEncodingException e) {
				log.error("Error", e);
			} catch (MalformedURLException e) {
				log.error("Error", e);
			} catch (IOException e) {
				log.error("Error", e);
			}			
		}
	}
}
